# -*- coding: UTF-8 -*-
"""
@File    ：案例2.py
@Author  ：techPang
@Date    ：2025/2/8 16:14 
@Version ：1.0
需求:
    1.各个城市销售额排名，从大到小
    2.全部城市，有哪些商品类别在售卖
    3.北京市有哪些商品类别在售卖
"""
from json import loads

from pyspark import SparkContext, SparkConf
# 1.构建执行环境入口对象
conf = SparkConf().setMaster("local[*]").setAppName("test_spark")
sc = SparkContext(conf=conf)
# 2.读取数据文件
rdd = sc.textFile("./orders.txt")

# TODO 需求1.各个城市销售额排名，从大到小
# 1.1取出一个个json
rdd_json = rdd.flatMap(lambda x: x.split("|"))
# print(rdd_json.collect())
# 1.2将json转为字典
rdd_dict = rdd_json.map(lambda x: loads(x))
# 1.3取出城市和销售额数据
rdd_tuple = rdd_dict.map(lambda x: (x.get("areaName"), int(x.get("money"))))
# 1.4按城市分组按销售额聚合
rdd_group = rdd_tuple.reduceByKey(lambda x, y: x + y)
# 1.5 按销售额聚合结果进行排序
result = rdd_group.sortBy(lambda x: x[1], ascending=False, numPartitions=1)
print(result.collect())
# TODO 需求2.全部城市，有哪些商品类别在售卖
rdd_all = rdd_dict.map(lambda x: x.get("category")).distinct()
print(rdd_all.collect())
# TODO 需求3.天津市有哪些商品类别在售卖
rdd_bj = rdd_dict.filter(lambda x: x.get("areaName") == "天津").map(lambda x: x.get("category")).distinct()
print(rdd_bj.collect())
# 停止运行
sc.stop()
